package com.wtw.graph

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.{SparkConf, SparkContext}

object GrapxDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("graphDemo").setMaster("local")
    val sc = new SparkContext(conf)

    val vertexArray = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )

    val edgeArray = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )


    val vertexs = sc.parallelize(vertexArray)
    val edges = sc.parallelize(edgeArray)

    //创建图对象
    val graph = Graph(vertexs, edges)

    //（2）找出图中年龄大于 30 的顶点

    graph.vertices.filter { case (id, (name, age)) => age > 30 }.collect().foreach(println(_))

    println("-------------图中年龄大于 30 的顶点----------")

    //    （3）找出图中属性大于 5 的边

    graph.edges.filter(e => e.attr > 5).collect().foreach(println(_))

    println("-------------图中属性大于 5 的边----------")

    //    （4）找出图中最大的出度、入度、度数

    def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
      if(a._2 > b._2) a else b
    }

    println("max of outDegrees:" + graph.outDegrees.reduce(max) +
      " max of inDegrees:" + graph.inDegrees.reduce(max) +
      " max of Degrees:" + graph.degrees.reduce(max))
  }
}
